
package com.hery.core;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.hery.config.ConfigFactory;



public class NettyServer extends AbstractEndpoint {
    
    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
    private Map<String, NettyChannel>  channels; // <ip:port, channel>

    private ServerBootstrap                 bootstrap;

    private io.netty.channel.Channel channel;

    private int port=9009;
    private ConfigFactory config;

    EventLoopGroup boss=new NioEventLoopGroup();
	EventLoopGroup worker=new NioEventLoopGroup();

    public NettyServer(int port) throws RemotingException{
    	
        super();
        setPort(port);
        try {
            doOpen();
            if (logger.isInfoEnabled()) {
                logger.info("Start " + getClass().getSimpleName() + " bind " + port );
            }
        } catch (Throwable t) {
            throw new RemotingException( "Failed to bind " + getClass().getSimpleName() 
                                        + " on " +port+ ", cause: " + t.getMessage(), t);
        }

    }

    protected void doOpen() throws Throwable {
    	EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			ServerBootstrap boot = new ServerBootstrap();
			boot.group(bossGroup, workerGroup);
			boot.channel(NioServerSocketChannel.class);
			final NettyHandler nettyHandler = new NettyHandler();
			boot.childHandler(new ChannelInitializer(){
				@Override
				protected void initChannel(io.netty.channel.Channel nChannel) throws Exception {
					ChannelPipeline pipeline= nChannel.pipeline();
					pipeline.addLast(new LengthFieldBasedFrameDecoder(100000000,0,4,0,4));
					pipeline.addLast(nettyHandler);
				}
	        	
	        });
			boot.option(ChannelOption.SO_BACKLOG, 128);
			boot.childOption(ChannelOption.SO_KEEPALIVE, true);
			ChannelFuture future = boot.bind(port).sync();
			future.channel().closeFuture().sync();
		} finally {
			workerGroup.shutdownGracefully();
			bossGroup.shutdownGracefully();
		}
    }

    protected void doClose() throws Throwable {
        try {
            if (channel != null) {
                // unbind.
                channel.close();
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
        try {
            Collection<NettyChannel> channels = getChannels();
            if (channels != null && channels.size() > 0) {
                for (NettyChannel channel : channels) {
                    try {
                        channel.close();
                    } catch (Throwable e) {
                        logger.warn(e.getMessage(), e);
                    }
                }
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
        
        try {
            if (channels != null) {
                channels.clear();
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
        try {
            if (bootstrap != null) { 
               boss.shutdownGracefully();
               worker.shutdownGracefully();
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
    }
    
    public Collection<NettyChannel> getChannels() {
        Collection<NettyChannel> chs = new HashSet<NettyChannel>();
        for (NettyChannel channel : this.channels.values()) {
            if (channel.isConnected()) {
                chs.add(channel);
            } else {
                channels.remove(toAddressString(channel.getRemoteAddress()));
            }
        }
        return chs;
    }

    public NettyChannel getChannel(InetSocketAddress remoteAddress) {
        return channels.get(toAddressString(remoteAddress));
    }
    private  String toAddressString(InetSocketAddress address) {
        return address.getAddress().getHostAddress() + ":" + address.getPort();
    }



 

}